package net.fruit.mq.client.impl.rocketmq;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;

import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;

import net.fruit.mq.client.Message;
import net.fruit.mq.client.MessageAccessor;
import net.fruit.mq.client.exception.LMQClientException;

public class RMQUtil {
    private static Set<String> ReservedKeySetRMQ = new HashSet<String>();

    private static Set<String> ReservedKeySetLOCAL = new HashSet<String>();

    static {

        /**
         * RMQ
         */
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_KEYS);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_TAGS);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_WAIT_STORE_MSG_OK);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_RETRY_TOPIC);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_REAL_TOPIC);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_REAL_QUEUE_ID);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_TRANSACTION_PREPARED);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_PRODUCER_GROUP);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_MIN_OFFSET);
        ReservedKeySetRMQ.add(MessageConst.PROPERTY_MAX_OFFSET);

        /**
         * LOCAL
         */
        ReservedKeySetLOCAL.add(Message.SystemPropKey.TAG);
        ReservedKeySetLOCAL.add(Message.SystemPropKey.KEY);
        ReservedKeySetLOCAL.add(Message.SystemPropKey.MSGID);
        ReservedKeySetLOCAL.add(Message.SystemPropKey.RECONSUMETIMES);
        ReservedKeySetLOCAL.add(Message.SystemPropKey.STARTDELIVERTIME);
    }


    public static org.apache.rocketmq.common.message.Message msgConvert(net.fruit.mq.client.Message message) {
        org.apache.rocketmq.common.message.Message msgRMQ = new org.apache.rocketmq.common.message.Message();
        if (message == null) {
            throw new LMQClientException("\'message\' is null");
        }

        if (message.getTopic() != null) {
            msgRMQ.setTopic(message.getTopic());
        }
        if (message.getKey() != null) {
            msgRMQ.setKeys(message.getKey());
        }
        if (message.getTag() != null) {
            msgRMQ.setTags(message.getTag());
        }
        if (message.getStartDeliverTime() > 0) {
            msgRMQ.putUserProperty(Message.SystemPropKey.STARTDELIVERTIME, String.valueOf(message.getStartDeliverTime()));
        }
        if (message.getBody() != null) {
            msgRMQ.setBody(message.getBody());
        }

        Properties systemProperties = MessageAccessor.getSystemProperties(message);
        if (systemProperties != null) {
            Iterator<Entry<Object, Object>> it = systemProperties.entrySet().iterator();
            while (it.hasNext()) {
                Entry<Object, Object> next = it.next();
                if (!ReservedKeySetLOCAL.contains(next.getKey().toString())) {
                    org.apache.rocketmq.common.message.MessageAccessor.putProperty(msgRMQ, next.getKey().toString(),
                            next.getValue().toString());
                }
            }
        }

        Properties userProperties = message.getUserProperties();
        if (userProperties != null) {
            Iterator<Entry<Object, Object>> it = userProperties.entrySet().iterator();
            while (it.hasNext()) {
                Entry<Object, Object> next = it.next();
                if (!ReservedKeySetRMQ.contains(next.getKey().toString())) {
                    org.apache.rocketmq.common.message.MessageAccessor.putProperty(msgRMQ, next.getKey().toString(),
                            next.getValue().toString());
                }
            }
        }

        return msgRMQ;
    }

    public static net.fruit.mq.client.Message msgConvert(org.apache.rocketmq.common.message.Message msgRMQ) {
        net.fruit.mq.client.Message message = new net.fruit.mq.client.Message();
        if (msgRMQ.getTopic() != null) {
            message.setTopic(msgRMQ.getTopic());
        }
        if (msgRMQ.getKeys() != null) {
            message.setKey(msgRMQ.getKeys());
        }
        if (msgRMQ.getTags() != null) {
            message.setTag(msgRMQ.getTags());
        }
        if (msgRMQ.getBody() != null) {
            message.setBody(msgRMQ.getBody());
        }

        message.setReconsumeTimes(((MessageExt) msgRMQ).getReconsumeTimes());

        Map<String, String> properties = msgRMQ.getProperties();
        if (properties != null) {
            Iterator<Entry<String, String>> it = properties.entrySet().iterator();
            while (it.hasNext()) {
                Entry<String, String> next = it.next();
                // System
                if (ReservedKeySetRMQ.contains(next.getKey().toString())) {
                    MessageAccessor.putSystemProperties(message, next.getKey().toString(), next.getValue().toString());
                }
                // User
                else {
                    message.putUserProperties(next.getKey().toString(), next.getValue().toString());
                }
            }
        }

        return message;
    }
}
